package com.kwan.xframe.mvp.model.api;

import android.util.Log;

import com.orhanobut.logger.Logger;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;


public class RxSocket2 {

	/*
	 *常量
	 **/
	private String TAG = "RxSocket2";
	private boolean OpenLog = false;
	private long CONNECT_TIME_OUT = 3000;

	/*
	 *单例
	 * */
	private Subject<byte[]> readSubject;
	private Subject<SocketStatus> connectStatus;

	private static volatile RxSocket2 defaultInstance;

	private RxSocket2() {
		readSubject = PublishSubject.<byte[]>create().toSerialized();
		connectStatus = PublishSubject.<SocketStatus>create().toSerialized();
	}

	public static RxSocket2 getInstance() {
		RxSocket2 rxSocket2 = defaultInstance;
		if (defaultInstance == null) {
			synchronized (RxSocket2.class) {
				rxSocket2 = defaultInstance;
				if (defaultInstance == null) {
					rxSocket2 = new RxSocket2();
					defaultInstance = rxSocket2;
				}
			}
		}
		return rxSocket2;
	}

	/*  变量
	 * */
	private SocketStatus socketStatus = SocketStatus.DIS_CONNECT;
	private Selector selector = null;
	private SocketChannel socketChannel = null;
	private SelectionKey selectionKey = null;
	private ReadThread readThread = null;
	private boolean isReadThreadAlive = true;
	private SocketReconnectCallback socketReconnectCallback = null;

	/*  方法
	 * */

	/**
	 * 监听Socket的状态
	 *
	 * @return Rx SocketStatus 状态
	 */
	public Observable<SocketStatus> socketStatusListener() {
		return connectStatus;
	}

	/**
	 * 建立Socket连接，只是尝试建立一次
	 *
	 * @param ip   IP or 域名
	 * @param port 端口
	 * @return Rx true or false
	 */
	public Flowable<Boolean> connectRx(final String ip, final int port) {
		return Flowable
				.create(new FlowableOnSubscribe<Boolean>() {
					@Override
					public void subscribe(final FlowableEmitter<Boolean> e) {
						//正在连接
						if (socketStatus == SocketStatus.CONNECTING) {
							e.onNext(false);
							e.onComplete();
							return;
						}
						//未连接 | 已经连接，关闭Socket
						socketStatus = SocketStatus.DIS_CONNECT;
						isReadThreadAlive = false;
						readThread = null;
						if (selector != null)
							try {
								selector.close();
							} catch (Exception ex) {
								//LogUtils.i(OpenLog, TAG, "selector.close");
							}
						if (selectionKey != null)
							try {
								selectionKey.cancel();
							} catch (Exception ex) {
								//LogUtils.i(OpenLog, TAG, "selectionKey.cancel");
							}
						if (socketChannel != null)
							try {
								socketChannel.close();
							} catch (Exception ex) {
								//LogUtils.i(OpenLog, TAG, "socketChannel.close");
							}

						//重启Socket
						isReadThreadAlive = true;
						readThread = new ReadThread(ip, port);
						readThread.start();
						socketReconnectCallback = new SocketReconnectCallback() {
							@Override
							public void onSuccess() {
								Logger.e( "connectRx:" + "CONNECTED");
								socketStatus = SocketStatus.CONNECTED;
								e.onNext(true);
								e.onComplete();
							}

							@Override
							public void onFail(String msg) {
								Logger.e("connectRx:" + msg);
								e.onNext(false);
								e.onComplete();
							}
						};
					}
				}, BackpressureStrategy.BUFFER).subscribeOn(Schedulers.newThread());
	}

	/**
	 * 断开当前的Socket
	 * 还能再继续连接
	 *
	 * @return Rx true or false
	 */
	public Flowable<Boolean> disConnect() {
		return Flowable.create(new FlowableOnSubscribe<Boolean>() {
			@Override
			public void subscribe(FlowableEmitter<Boolean> e) throws Exception {
				try {
					if (socketStatus == SocketStatus.DIS_CONNECT) {
						e.onNext(true);
						e.onComplete();
					} else {
						socketStatus = SocketStatus.DIS_CONNECT;
						isReadThreadAlive = false;
						readThread = null;
						if (selector != null)
							try {
								selector.close();
							} catch (Exception ex) {
								//LogUtils.i(OpenLog, TAG, "selector.close");
							}
						if (selectionKey != null)
							try {
								selectionKey.cancel();
							} catch (Exception ex) {
								//LogUtils.i(OpenLog, TAG, "selectionKey.cancel");
							}
						if (socketChannel != null)
							try {
								socketChannel.close();
							} catch (Exception ex) {
								//	LogUtils.i(OpenLog, TAG, "socketChannel.close");
							}
						e.onNext(true);
						e.onComplete();
					}
				} catch (Exception ex) {
					e.onNext(false);
					e.onComplete();
				}
			}
		}, BackpressureStrategy.BUFFER);
	}

	/**
	 * 读取Socket的消息
	 *
	 * @return Rx error 或者 有数据
	 */
	public Observable<byte[]> read() {
		return readSubject;
	}

	/**
	 * 向Socket写消息
	 *
	 * @param buffer 数据包
	 * @return Rx true or false
	 */
	public Flowable<Boolean> write(final ByteBuffer buffer) {
		long WRITE_TIME_OUT = 3000;
		return Flowable
				.create(new FlowableOnSubscribe<Boolean>() {
					@Override
					public void subscribe(FlowableEmitter<Boolean> e) {
						if (socketStatus != SocketStatus.CONNECTED) {
							//LogUtils.i(OpenLog, TAG, "write." + "SocketStatus.DISCONNECTED");
							e.onNext(false);
							e.onComplete();
						} else {
							if (socketChannel != null && socketChannel.isConnected()) {
								try {
									int result = socketChannel.write(buffer);
									if (result < 0) {
										//	LogUtils.i(OpenLog, TAG, "write." + "发送出错");
										e.onNext(false);
										e.onComplete();
									} else {
										//	LogUtils.i(OpenLog, TAG, "write." + "success!");
										e.onNext(true);
										e.onComplete();
									}
								} catch (Exception ex) {
									//LogUtils.i(OpenLog,TAG,"write."+e.getMessage());
									e.onNext(false);
									e.onComplete();
								}
							} else {
								//	LogUtils.i(OpenLog,TAG,"write."+"close");
								e.onNext(false);
								e.onComplete();
							}
						}

					}
				}, BackpressureStrategy.BUFFER).subscribeOn(Schedulers.newThread()).
						timeout(WRITE_TIME_OUT, TimeUnit.MILLISECONDS, Flowable.just(false));

	}


	/**
	 * 获取Socket的链接状态
	 *
	 * @return 状态
	 */
	public SocketStatus getSocketStatus() {
		return socketStatus;
	}

	//
	/*  类 && 枚举 && 接口
	 * */
	private class ReadThread extends Thread {
		private String ip;
		private int port;

		ReadThread(String ip, int port) {
			this.ip = ip;
			this.port = port;
		}

		@Override
		public void run() {
			Logger.e("ReadThread:" + "start");
			while (isReadThreadAlive) {
				//连接
				if (socketStatus == SocketStatus.DIS_CONNECT) {
					try {
						Logger.e("ReadThread:" + "DIS_CONNECT");
						if (selectionKey != null) selectionKey.cancel();
						socketChannel = SocketChannel.open();
						socketChannel.configureBlocking(false);
						selector = Selector.open();

						socketChannel.connect(new InetSocketAddress(ip, port));
						selectionKey = socketChannel.register(selector, SelectionKey.OP_CONNECT);

						socketStatus = SocketStatus.CONNECTING;
						connectStatus.onNext(SocketStatus.CONNECTING);
					}


					catch (Exception e) {
						isReadThreadAlive = false;
						socketStatus = SocketStatus.DIS_CONNECT;
						connectStatus.onNext(SocketStatus.DIS_CONNECT);
						Logger.e("ReadThread:init:" + e.getMessage());
						if (socketReconnectCallback != null)
							socketReconnectCallback.onFail("SocketConnectFail1");
					}
				} else if (socketStatus == SocketStatus.CONNECTING || socketStatus == SocketStatus.CONNECTED) {
					try {

						Log.e("kwan","xx" +(selector==null));

						selector.select();
						Iterator<SelectionKey> it = selector.selectedKeys().iterator();
						while (it.hasNext()) {
							SelectionKey key = it.next();
							if (key.isConnectable()) {
								if (socketChannel.isConnectionPending()) {
									try {

										Log.e("kwan","1");

										//Log.e("kwan","1"+socketChannel.isConnected());
										//socketChannel.finishConnect();
										Log.e("kwan","2");
										socketStatus = SocketStatus.CONNECTED;
										connectStatus.onNext(SocketStatus.CONNECTED);
										socketChannel.configureBlocking(false);
										Log.e("kwan","3");
										socketChannel.register(selector, SelectionKey.OP_READ);
										Log.e("kwan","4");
										if (socketReconnectCallback != null)
											socketReconnectCallback.onSuccess();
									} catch (Exception e) {
										isReadThreadAlive = false;
										socketStatus = SocketStatus.DIS_CONNECT;
										connectStatus.onNext(SocketStatus.DIS_CONNECT);
										Logger.e("ReadThread:finish:" + e.getLocalizedMessage());
										if (socketReconnectCallback != null)
											socketReconnectCallback.onFail("SocketConnectFail2");
									}
								}
							} else if (key.isReadable()) {
								ByteBuffer buf = ByteBuffer.allocate(1000);

								Log.e("kwan","rr::"+(socketChannel==null));
								Log.e("kwan","rr::"+(buf==null));
								int length = socketChannel.read(buf);

								Log.e("kwan","rrrrrrr::"+(buf==null));

								if (length <= 0) {
									Logger.e("服务器主动断开链接！");
									isReadThreadAlive = false;
									socketStatus = SocketStatus.DIS_CONNECT;
									connectStatus.onNext(SocketStatus.DIS_CONNECT);
									if (socketReconnectCallback != null)
										socketReconnectCallback.onFail("SocketConnectFail3");
								} else {
									Logger.e("readSubject:msg！" + "length:" + length);
									byte[] bytes = new byte[length];
									for (int i = 0; i < length; i++) {
										bytes[i] = buf.get(i);
									}
									readSubject.onNext(bytes);
								}
							}
						}

						Log.e("kwan","to move"+(it==null));


						it.remove();
					} catch (Exception e) {
						isReadThreadAlive = false;
						socketStatus = SocketStatus.DIS_CONNECT;
						connectStatus.onNext(SocketStatus.DIS_CONNECT);
						Logger.e("ReadThread:read:" + e.getMessage());
						if (socketReconnectCallback != null)
							socketReconnectCallback.onFail("SocketConnectFail4");
					}
				}
			}
		}
	}

	public enum SocketStatus {
		DIS_CONNECT,
		CONNECTING,
		CONNECTED,
	}

	private interface SocketReconnectCallback {
		void onSuccess();

		void onFail(String msg);
	}
}


